/*
 * Copyright 2020-2025 the original author or authors.
 * You cannot use this file unless authorized by the author.
 */
package org.ipig.data.as.impl;

import org.ipig.commons.conf.spike.SpikeConf;
import org.ipig.data.as.SpikeService;
import com.aerospike.client.AerospikeClient;
import com.aerospike.client.async.EventLoops;
import com.aerospike.client.async.EventPolicy;
import com.aerospike.client.async.NettyEventLoops;
import com.aerospike.client.policy.ClientPolicy;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;

/**
 * SpikeManager
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id: SpikeManager.java 3 2020-01-21 09:36:33Z ts <comchnts@163.com> $$
 * @since 1.0
 */
@Slf4j
public class SpikeManager implements SpikeService<AerospikeClient>, KryoSerializable, Serializable {
    private static final long serialVersionUID = 5413574768077033386L;
    @Getter
    @Setter
    private SpikeConf conf;
    private transient EventLoops eventLoops = null;
    private volatile transient AerospikeClient client = null;

    public SpikeManager() {
    }

    /**
     * construct
     *
     * @param conf
     */
    public SpikeManager(SpikeConf conf) {
        this.setConf(conf);
    }

    @Override
    public AerospikeClient getClient() {
        if (null == client) {
            synchronized (this) {
                if (null == client) {
                    if (conf == null) {
                        new IllegalArgumentException("未设置SpikeConf对象");
                    }
                    ClientPolicy clientPolicy = new ClientPolicy();
                    if (eventLoops == null) {
                        EventPolicy eventPolicy = new EventPolicy();
                        EventLoopGroup group = new NioEventLoopGroup();
                        eventLoops = new NettyEventLoops(eventPolicy, group);
                    }
                    conf.eventLoops = eventLoops;
                    client = new AerospikeClient(conf, conf.getHosts());
                    Runtime.getRuntime().addShutdownHook(new Thread() {
                        @Override
                        public void run() {
                            close();
                        }
                    });
                }
            }
        }
        return client;
    }


    public void close() {
        if (eventLoops != null) {
            eventLoops.close();
        }
        if (client != null) {
            client.close();
        }
    }

    public void write(Kryo kryo, Output output) {
        log.info("序列化");
        kryo.writeObject(output, this.getConf());
    }

    public void read(Kryo kryo, Input input) {
        log.info("反序列化，同时构建SpikeClient");
        conf = kryo.readObject(input, SpikeConf.class);
    }
}